package com.binance.assetservice.web.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.binance.assetservice.data.models.ProductItem;
import com.binance.assetservice.web.service.IOrderBookManagerService;
import com.binance.assetservice.web.service.IOrderBookService;
import com.binance.assetservice.web.service.IProductService;
import com.binance.assetservice.web.utils.WebsocketClientEndpoint;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import javax.annotation.PostConstruct;
import javax.websocket.CloseReason;
import javax.websocket.Session;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

@Service
public class OrderBookManagerServiceImpl implements IOrderBookManagerService {

    final public static long NINE_O_CLOCK = 60 * 60 * 1000 * 9;

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    final private Map<String, IOrderBookService> cache = Maps.newHashMap();
    private final BlockingQueue<JSONArray> updatesQueue = new LinkedBlockingQueue<JSONArray>();
    @Autowired
    private RestTemplate template;
    @Value("${matchbox.rest.root.url}")
    private String restUrl;
    @Value("${matchbox.management.root.url}")
    private String mgmtUrl;
    @Value("${matchbox.wss.internal.root.url}")
    private String wssUrl;
    @Value("${matchbox.kline.internal.root.url}")
    private String klineUrl;
    @Autowired
    private IProductService productService;
    private WebsocketClientEndpoint clientEndPoint;
    private volatile Long refreshKlineTime = 0L;

    @PostConstruct
    public void init() throws Exception {
        List<ProductItem> productList = this.productService.getAllProducts();
        for (ProductItem item : productList) {
            cache.put(item.getSymbol(), new OrderBookServiceImpl(item.getSymbol(), template, restUrl));
        }
        try {
            this.initClientEndPoint();
            new Thread(new Runnable() {
                @Override
                public void run() {
                    if (clientEndPoint != null) {
                        clientEndPoint.connect();
                    }
                }
            }).start();

            final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("stream-update-%d").build();
            final ExecutorService es = Executors.newSingleThreadExecutor(threadFactory);
            es.submit(new Runnable() {

                @Override
                public void run() {
                    while (true) {
                        JSONArray obj = null;
                        try {
                                obj = updatesQueue.take();
                                if (null != obj)
                                    processUpdate(obj);
                        } catch (Exception e) {
                            logger.error(String.format("OrderBook failed to process update: %s", obj), e);
                        }
                    }

                }
            });

        } catch (Exception ex) {
            logger.error("OrderBook failed to connect to websocket", ex);
        }

        // 定时刷新K线
        final Runnable connectTask = new Runnable() {

            @Override
            public void run() {
                try {
                    connectWss();
                } catch (Exception e) {
                    logger.error("Reconnect to process failed:", e);
                }
            }
        };
        Executors.newScheduledThreadPool(1).scheduleWithFixedDelay(connectTask, 0L, 60L, TimeUnit.SECONDS);
    }


    private void initClientEndPoint() throws URISyntaxException {
        final String wsUrl = String.format("%s/!miniTicker@arr", wssUrl);
        clientEndPoint = new WebsocketClientEndpoint(new URI(wsUrl));
        clientEndPoint.addMessageHandler(new WebsocketClientEndpoint.MessageHandler() {
            @Override
            public void handleMessage(String message) {
                try {
                    final JSONArray obj = JSON.parseArray(message);
                    updatesQueue.put(obj);
                    refreshKlineTime = System.currentTimeMillis();
                } catch (Exception e) {
                    logger.error(String.format("OrderBook failed to process market data update: %s", message), e);
                }
            }

            @Override
            public void handleOpenEvent(Session session) {}

            @Override
            public void handleCloseEvent(Session session, CloseReason reason) {
                if (null != clientEndPoint) {
                    try {
                        clientEndPoint.connect();
                    } catch (Exception e) {
                        logger.error(String.format("OrderBook failed to connect to %s", wsUrl), e);
                    }
                }
            }
        });
    }

    private void connectWss() {
        logger.info("OrderBook last refreshTime:" + refreshKlineTime);
        if (System.currentTimeMillis() - refreshKlineTime > 60 * 1000L) {
            try {
                if (clientEndPoint == null) {
                    this.initClientEndPoint();
                }
                clientEndPoint.close();
                clientEndPoint.connect();
            } catch (Exception e) {
                logger.error("Close or Reconnect to process failed:", e);
            }
        }
    }

    private void processUpdate(JSONArray obj) {
        for (int i = 0; i < obj.size(); i++) {
            JSONObject jsonObj = (JSONObject) obj.get(i);
            IOrderBookService orderBook = cache.get(jsonObj.getString("s"));
            if (orderBook != null) {
                orderBook.setKlineT(jsonObj);
            }
        }
    }

    @Override
    public void reInit() throws Exception {
        List<ProductItem> productList = this.productService.getAllProducts();
        for (ProductItem item : productList) {
            cache.put(item.getSymbol(), new OrderBookServiceImpl(item.getSymbol(), template, restUrl));
        }
    }

    @Override
    public Long getLastAggTradeId(String symbol) {
        if (!cache.containsKey(symbol))
            return -1L;
        return cache.get(symbol).getLastAggTradeId();
    }

    @Override
    public JSONArray getTodayTrades(String symbol) {
        if (!cache.containsKey(symbol))
            return null;
        return cache.get(symbol).getTodayTrades();
    }

    @Override
    public double getActiveBuy(String symbol) {
        if (!cache.containsKey(symbol))
            return 0L;
        return cache.get(symbol).getActiveBuy();
    }

    @Override
    public double getActiveSell(String symbol) {
        if (!cache.containsKey(symbol))
            return 0L;
        return cache.get(symbol).getActiveSell();
    }

    @Override
    public JSONObject getTicker(String symbol) {
        if (!cache.containsKey(symbol))
            return null;
        return cache.get(symbol).getTicker();
    }

    @Override
    public Double getVolume(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getVolume();
    }

    @Override
    public Double getTurnOver(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getTurnOver();
    }

    @Override
    public Double getOpen(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getOpen();
    }

    @Override
    public Double getLow(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getLow();
    }

    @Override
    public Double getHigh(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getHigh();
    }

    @Override
    public Double getClose(String symbol) {
        if (!cache.containsKey(symbol))
            return 0d;
        return cache.get(symbol).getClose();
    }

    @Override
    public Long getRefreshTime() {
        return refreshKlineTime;
    }

}
